package core

import (
	"encoding/json"
	"gark-elastic-job/common"
	"gark-elastic-job/engine"
	logger "github.com/sirupsen/logrus"
	"sync"
)

const jobExecuteLogPath = "/gark0/client/execute/log/"

type Collector struct {
	jobNode   *JobNode
	saver     Saver
	locker    *sync.RWMutex
	watchPath string
}

func NewCollector(node *JobNode, saver Saver) *Collector {
	logger.Debug("start Job Collector")
	collector := &Collector{
		jobNode:   node,
		watchPath: jobExecuteLogPath,
		saver:     saver,
		locker:    &sync.RWMutex{},
	}
	go collector.watchExecutionLogPath()
	go collector.loop()
	return collector
}

func (c *Collector) watchExecutionLogPath() {
	keyChangeResponse := c.jobNode.etcd.WatchKeyWithPrefix(c.watchPath)
	for event := range keyChangeResponse.Event {
		logger.Debugf("execution log path:%s do changed: %v", event.Key, event.Value)
		c.handleExecutionLogCollectionEvent(event)
	}

}

func (c *Collector) handleExecutionLogCollectionEvent(event *engine.KeyChangeEvent) {
	// follower node has no auth to write job execution log into storage system
	if c.jobNode.State == engine.Follower {
		return
	}

	switch event.Type {
	case engine.KeyCreateChangeEvent:
		if len(event.Value) == 0 {
			return
		}
		executionLog := &common.JobExecutionLog{}
		err := json.Unmarshal(event.Value, executionLog)
		if err != nil {
			logger.Errorf("[create]failed to decode execution log: %v, err: %v", event.Value, err)
			_ = c.jobNode.etcd.DeleteKey(event.Key)
			return
		}
		c.handleExecutionLogEvent(event.Key, executionLog)

	case engine.KeyUpdateChangeEvent:
		if len(event.Value) == 0 {
			return
		}
		executionLog := &common.JobExecutionLog{}
		err := json.Unmarshal(event.Value, executionLog)
		if err != nil {
			logger.Errorf("[update]failed to decode execution log: %v, err: %v", event.Value, err)
			return
		}
		c.handleExecutionLogEvent(event.Key, executionLog)

	case engine.KeyDeleteChangeEvent:
		// ignore
	}

}

func (c *Collector) loop() {
}

func (c *Collector) handleExecutionLogEvent(key string, executionLog *common.JobExecutionLog) {
	c.locker.Lock()
	defer c.locker.Unlock()

	exist, err := c.saver.CheckExist(executionLog.ID)
	if err != nil {
		logger.Errorf("failed to check execution log:%s exist, err:%v", executionLog.ID, err)
		return
	}

	if exist {
		c.saver.Save(*executionLog)
	} else {
		c.saver.Update(*executionLog)
	}
}
